#include "message_distribute_task.h"

#include "im/adapters/connection_manager_adapter.h"
#include "im/adapters/sock_message_outlet_adapter.h"
#include "im/adapters/group_adapter.h"
#include "im/adapters/push_adapter.h"
#include "base/curl_client.h"
#include "base/url_encode.h"
#include "base/hash.h"
//#include "im/adapters/comet_message_outlet_adapter.h"

namespace imserver {

void MessageDistributeTask::run() {
  LOG_INFO("MessageDistributeTask::run --> try to distribute msg. msg_id:" << message_.id << " type:" <<
      message_.type << " from:" << message_.sender << " to:" << message_.receiver << " did:" << message_.dialogue_id);
  if (message_.type == ImMessageType::IM_MESSAGE_P2P ||
      message_.type == ImMessageType::IM_MESSAGE_RESUME ||
      message_.type == ImMessageType::IM_MESSAGE_TYPING ||
      message_.type == ImMessageType::IM_MESSAGE_RESUME_LIKE ||
      message_.type == ImMessageType::IM_MESSAGE_RESUME_IGNORE ||
      message_.type == ImMessageType::IM_MESSAGE_JOB_INVITATION) {
    std::set<int64_t> uids;
    uids.insert(message_.receiver);
    std::map<int64_t, std::string> uid_to_sock_server_key;
    ConnectionManagerAdapter::instance().GetSockServiceKeyByUids(uid_to_sock_server_key, uids, 2);
    if (uid_to_sock_server_key.find(message_.receiver) == uid_to_sock_server_key.end() ||
        uid_to_sock_server_key[message_.receiver].find(":ok") == std::string::npos) {
        if (message_.type != ImMessageType::IM_MESSAGE_TYPING) {
          LOG_INFO("MessageDistributeTask::run --> push msg. msg_id:" << message_.id << " type:" << message_.type <<
              " from:" << message_.sender << " to:" << message_.receiver << " did:" << message_.dialogue_id);
          PushAdapter::instance().Push(message_.receiver, "您有一条新的消息！", message_.timestamp);
        }
    } else {
      std::string server_key = uid_to_sock_server_key[message_.receiver].
          substr(0, uid_to_sock_server_key[message_.receiver].find(":"));
      LOG_INFO("MessageDistributeTask::run --> send msg. server_key:" << server_key << " msg_id:" << message_.id <<
          " type:" << message_.type << " from:" << message_.sender << " to:" << message_.receiver <<
          " did:" << message_.dialogue_id);
      SockMessageOutletAdapter::instance().DistributeMessage(server_key, uids, message_, 2);
    }

    std::map<int64_t, std::string> uid_to_comet_server_key
        /*= ConnectionManagerAdapter::instance().GetCometServiceKeyByUids(uids, 2)*/;
    if (uid_to_comet_server_key.find(message_.receiver) != uid_to_comet_server_key.end() && 
        uid_to_comet_server_key[message_.receiver].find(":ok") != std::string::npos) {
      std::string server_key = uid_to_comet_server_key[message_.receiver].
          substr(0, uid_to_comet_server_key[message_.receiver].find(":"));
      //CometMessageOutletAdapter::instance().DistributeMessage(server_key, uids, message_, 2);
    }
  } else if (message_.type == ImMessageType::IM_MESSAGE_GROUP ||
             message_.type == ImMessageType::IM_MESSAGE_GROUP_JOIN ||
             message_.type == ImMessageType::IM_MESSAGE_GROUP_RESUME_LIKE ||
             message_.type == ImMessageType::IM_MESSAGE_GROUP_JOB_INVITATION) {
    std::set<int64_t> uids; 
    GroupAdapter::instance().GetGroupMemberIds(uids, message_.group_id);
    LOG_INFO("MessageDistributeTask::run --> get uids from group service. group_id:" << message_.group_id <<
        " uids_count:" << uids.size());
    if (message_.type != ImMessageType::IM_MESSAGE_GROUP_RESUME_LIKE) {
      uids.erase(message_.sender);
    }
    std::map<int64_t, std::string> uid_to_sock_server_key;
    ConnectionManagerAdapter::instance().GetSockServiceKeyByUids(uid_to_sock_server_key, uids, 2);
    std::map<int64_t, std::string> uid_to_comet_server_key
        /*= ConnectionManagerAdapter::instance().GetCometServiceKeyByUids(uids, 2)*/;

    std::set<int64_t> uids_notify;
    std::map<std::string, std::set<int64_t> > sock_server_key_2_uids;
    std::map<std::string, std::set<int64_t> > comet_server_key_2_uids;
    BOOST_AUTO(it1, uids.begin());
    for (; it1 != uids.end(); ++it1) {
      if (uid_to_sock_server_key.find(*it1) == uid_to_sock_server_key.end() || 
          uid_to_sock_server_key[*it1].find(":ok") == std::string::npos) {
        uids_notify.insert(*it1);
      } else {
        std::string server_key = uid_to_sock_server_key[*it1].substr(0, uid_to_sock_server_key[*it1].find(":"));
        if (sock_server_key_2_uids.find(server_key) != sock_server_key_2_uids.end()) {
          sock_server_key_2_uids[server_key].insert(*it1);
        } else {
          std::set<int64_t> uids_t;
          uids_t.insert(*it1);
          sock_server_key_2_uids.insert(make_pair(server_key, uids_t));
        }
      }

      if (uid_to_comet_server_key.find(*it1) != uid_to_comet_server_key.end() &&
          uid_to_comet_server_key[*it1].find(":ok") != std::string::npos) {

        std::string server_key = uid_to_comet_server_key[*it1].substr(0, uid_to_comet_server_key[*it1].find(":"));
        if (comet_server_key_2_uids.find(server_key) != comet_server_key_2_uids.end()) {
          comet_server_key_2_uids[server_key].insert(*it1);
        } else {
          std::set<int64_t> uids_t;
          uids_t.insert(*it1);
          comet_server_key_2_uids.insert(make_pair(server_key, uids_t));
        }
      }
    }

    BOOST_AUTO(it2, sock_server_key_2_uids.begin());
    for (; it2 != sock_server_key_2_uids.end(); ++it2) {
      LOG_INFO("MessageDistributeTask::run --> DistributeMessage. server_key:" << it2->first << " uids_count:"
          << (it2->second).size() << " msg_id:" << message_.id << " type:" << message_.type);
      SockMessageOutletAdapter::instance().DistributeMessage(it2->first, it2->second, message_, 2);
    }
    BOOST_AUTO(it3, comet_server_key_2_uids.begin());
    for (; it3 != comet_server_key_2_uids.end(); ++it3) {
      //CometMessageOutletAdapter::instance().DistributeMessage(it3->first, it3->second, message_, 2);
    }
  }
}

}

